package net.onething.dc.hdfs2bos;

import com.baidubce.services.bos.BosClient;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;

public class Main {
    static final Logger LOG = LoggerFactory.getLogger(Main.class);

    public static BosClient bosClient = null;
    public static BlockingQueue<FileToBosTask> taskBlockingQueue = null;
    public static ExecutorService executorPool = null;

    public static List<HdfsToBosExecutor> executors =
            new LinkedList<HdfsToBosExecutor>();

    public static void main(String[] args) {
        CommandLineParser parser = new DefaultParser();
        CommandLine cli;
        try {
            cli = parser.parse(OptionsArgsName.getAllSupportOption(), args);
            if (cli.hasOption(OptionsArgsName.HELP)) {
                OptionsArgsName.printHelpOption();
                return;
            }
        } catch (ParseException exp) {
            System.err.println("Parsing Argument failed. Reason: " + exp.getMessage());
            OptionsArgsName.printHelpOption();
            return;
        }

        ConfigReader configReader = new ConfigReader(cli);
        if (!configReader.isInitConfigFlag()) {
            System.err.println(configReader.getInitErrMsg());
            return;
        }
        if (null == Main.bosClient) {
            Main.bosClient = Utils.getBosClient(configReader);
        }
        if (null == Main.executorPool) {
            Main.executorPool = Executors.newFixedThreadPool(configReader.getMaxTaskNum());
        }
        if (null == Main.taskBlockingQueue) {
            Main.taskBlockingQueue = new LinkedBlockingQueue<FileToBosTask>(configReader.getMaxTaskNum() * 2);
        }

        Statistics.INSTANCE.start();
        // 启动消费者
        for (int i = 0; i < configReader.getMaxTaskNum(); i++) {
            HdfsToBosExecutor executor = new HdfsToBosExecutor(Main.taskBlockingQueue, true);
            Main.executorPool.submit(executor);
            Main.executors.add(executor);
        }
        Main.executorPool.shutdown();                // 停止提交新的任务
        HdfsToBos hdfsToBos = new HdfsToBos(configReader, Main.taskBlockingQueue, Main.bosClient);
        hdfsToBos.run();
        while (Main.taskBlockingQueue.size() != 0) {
            try {
                Thread.sleep(1 * 1000);
            } catch (InterruptedException e) {
                LOG.error("An exception occurred during the polling interval"
                        , e);
            }
        }
        for (HdfsToBosExecutor executor : Main.executors) {
            executor.stop();
        }
        try {
            Main.executorPool.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LOG.error("An exception occurred while waiting for the thread "
                    + "pool to complete", e);
        }
        Statistics.INSTANCE.printStatics();
        Main.bosClient.shutdown();
    }
}
